package worker

// ============================================================================
// Worker 模块 - 数据采集工作器
// ============================================================================
//
// 架构设计说明：
//
// Worker 负责 **主动轮询采集**，即网关主动向设备发送命令并等待响应。
//
// 支持的采集器类型：
//   - Serial (串口)：RS485、RS232 设备，Worker 管理串口连接
//   - TcpClient (TCP客户端)：网关主动连接设备，Worker 管理连接
//   - TcpServer (TCP服务端)：设备主动连接网关，连接由 zdm.go 管理，Worker 使用已建立的连接
//   - MQTT：通过 MQTT 协议通信
//
// TcpServer 的两种使用模式：
//   1. 纯主动上报：设备只上报数据，不需要 Worker（如 4G 水表、电表）
//      数据流程：设备 -> Listener(zdm.go) -> HandleActiveReport(Lua) -> 数据库
//   2. 轮询+上报：设备保持连接，Worker 定期发送命令查询，设备也可主动上报
//      数据流程：Worker -> TcpServerCollector -> 设备 -> Worker -> 数据库
//                设备 -> Listener(zdm.go) -> HandleActiveReport(Lua) -> 数据库（并行）
//
// Worker 的工作流程：
//   1. 定期调度设备采集任务
//   2. 调用 Lua 脚本生成命令（GenerateGetRealVariables / DeviceCustomCmd）
//   3. 通过采集器发送命令到设备（TcpServer 使用已建立的连接）
//   4. 接收并解析设备响应（AnalysisRx）
//   5. 更新设备状态到数据库
//   6. 触发事件通知
//
// ============================================================================

import (
	"encoding/json"
	"fmt"
	"strings"
	"time"
	"tinyGW/app/api/repository"
	"tinyGW/app/models"
	"tinyGW/pkg/service/collect/channel"
	"tinyGW/pkg/service/collect/collector"
	"tinyGW/pkg/service/command"
	"tinyGW/pkg/service/conf"
	"tinyGW/pkg/service/event"
	"tinyGW/pkg/service/script"

	"go.uber.org/zap"
)

type Worker interface {
	Start()
	Stop()
	CommandTask(task any)    // 指派高优先级任务（RPC 命令）
	CollectTask(task any)    // 指派正常任务（定时采集）
	CollectTaskIsFull() bool // 采集通道是否为空
}

var _ Worker = (*worker)(nil)

type worker struct {
	channel.PriorityChannel
	collector.Collector
	script.Runner
	deviceTypeRepository repository.DeviceTypeRepository
	deviceRepository     repository.DeviceRepository
	dataChan             chan []byte
	stopChan             chan int
	config               *conf.Config
	eventBus             *event.EventService
	commandManager       *command.Manager
}

// DataResult 数据接收结果
type DataResult struct {
	Success    bool
	Timeout    bool
	Data       []byte
	DataLength int
	Variables  []models.DeviceProperty
}

// DeviceCollectContext 设备采集上下文
type DeviceCollectContext struct {
	Device        *models.Device
	DeviceCopy    *models.Device
	DriverName    string
	Step          int
	IsCommand     bool
	CommandName   string
	CommandParams string
}

func NewWorker(
	domainCollector models.Collector,
	deviceRepository repository.DeviceRepository,
	config *conf.Config,
	eventBus *event.EventService,
	deviceTypeRepository repository.DeviceTypeRepository,
	commandManager *command.Manager,
) Worker {
	result := &worker{
		PriorityChannel:      channel.NewPriorityChannel(),
		Collector:            collector.ConnectorFactory(domainCollector, commandManager),
		Runner:               script.NewLuaRunner(),
		deviceRepository:     deviceRepository,
		dataChan:             make(chan []byte, 1024),
		stopChan:             make(chan int, 1),
		config:               config,
		eventBus:             eventBus,
		deviceTypeRepository: deviceTypeRepository,
		commandManager:       commandManager,
	}
	zap.S().Info("创建数据采集线程！", result.Collector)
	// RPC 命令优先级高
	result.PriorityChannel.SetPriorWorker(result.CommandExecutor)
	// 采集任务优先级低
	result.PriorityChannel.SetNormalWorker(result.CollectExecutor)

	return result
}

func (w *worker) CommandTask(task any) {
	w.PriorityChannel.DispatchPriorTask(task)
}

func (w *worker) CollectTask(task any) {
	w.PriorityChannel.DispatchNormalTask(task)
}

func (w *worker) CollectTaskIsFull() bool {
	return w.PriorityChannel.NormalTaskIsFull()
}

func (w *worker) Start() {
	// 如果不是每次都打开，那么只在这里打开一次
	if !w.config.Serial.OpenEveryTime {
		w.Collector.Open(nil)
	}

	go w.BlockRead()
	w.PriorityChannel.Start()
}

func (w *worker) Stop() {
	w.PriorityChannel.Stop()
	w.stopChan <- 1

	// 如果没事每次都打开，那么只在这里关闭一次
	if !w.config.Serial.OpenEveryTime {
		w.Collector.Close()
	}
}

// BlockRead 阻塞读取数据
func (w *worker) BlockRead() {
	// 预分配单个缓冲区并重用，避免频繁内存分配
	buf := make([]byte, 1024)

	// 自适应休眠时间
	sleepTime := 50 * time.Millisecond
	emptyReadCount := 0

	for {
		select {
		case <-w.stopChan:
			zap.S().Info("串口数据读取线程正确退出！")
			return
		default:
			count := w.Collector.Read(buf)

			if count > 0 {
				// 有数据读取时，重置空读计数和休眠时间
				dataCopy := make([]byte, count)
				copy(dataCopy, buf[:count])

				// 避免阻塞，使用非阻塞发送，如果通道满则记录日志
				select {
				case w.dataChan <- dataCopy:
					// 发送成功
					emptyReadCount = 0
					sleepTime = 50 * time.Millisecond
				default:
					// 通道已满，记录警告但不阻塞
					zap.S().Warnf("数据通道已满，丢弃读取的%d字节数据", count)
				}
			} else {
				// 无数据读取时，增加空读计数
				emptyReadCount++

				// 动态调整休眠时间，减少CPU使用
				if emptyReadCount > 20 {
					sleepTime = 200 * time.Millisecond
				}
				if emptyReadCount > 50 {
					sleepTime = 500 * time.Millisecond
				}
			}

			// 使用动态休眠时间
			time.Sleep(sleepTime)
		}
	}
}

// applyDeviceScale 应用设备倍率设置
func (w *worker) applyDeviceScale(device *models.Device) {
	if device.Scale > 1 {
		for k, p := range device.Type.Properties {
			if p.Name == "dev_consumption" || p.Name == "dev_flow" {
				device.Type.Properties[k].Scale = device.Scale
			}
		}
	}
}

// receiveData 接收并处理设备数据
func (w *worker) receiveData(ctx *DeviceCollectContext) *DataResult {
	rxTotalBuf := make([]byte, 0, 1024) // 预分配容量
	rxTotalBufCnt := 0

	for {
		select {
		case rxBuf := <-w.dataChan:
			rxBufCnt := len(rxBuf)
			if rxBufCnt > 0 {
				rxTotalBufCnt += rxBufCnt
				rxTotalBuf = append(rxTotalBuf, rxBuf...)
			}

			if rxTotalBufCnt > 0 {
				logMsg := fmt.Sprintf("从设备[%s]接收数据[%d:%X]", ctx.Device.Name, rxTotalBufCnt, rxTotalBuf)
				if ctx.IsCommand {
					zap.S().Infof("RPC%s", logMsg)
				} else {
					zap.S().Info(logMsg)
				}

				// 应用设备倍率
				w.applyDeviceScale(ctx.Device)

				var tempVariables []models.DeviceProperty
				if w.Runner.AnalysisRx(ctx.Device.Address, ctx.Device.Type.Properties, rxTotalBuf, rxTotalBufCnt, &tempVariables) {
					return &DataResult{
						Success:    true,
						Timeout:    false,
						Data:       rxTotalBuf,
						DataLength: rxTotalBufCnt,
						Variables:  tempVariables,
					}
				}
			}

		case <-time.After(time.Duration(w.Collector.GetTimeout()) * time.Millisecond):
			if rxTotalBufCnt > 0 {
				logMsg := fmt.Sprintf("[超时]从设备[%s]接收数据[%d:%X]", ctx.Device.Name, rxTotalBufCnt, rxTotalBuf)
				zap.S().Info(logMsg)

				// 应用设备倍率
				w.applyDeviceScale(ctx.Device)

				var tempVariables []models.DeviceProperty
				if w.Runner.AnalysisRx(ctx.Device.Address, ctx.Device.Type.Properties, rxTotalBuf, rxTotalBufCnt, &tempVariables) {
					return &DataResult{
						Success:    true,
						Timeout:    true,
						Data:       rxTotalBuf,
						DataLength: rxTotalBufCnt,
						Variables:  tempVariables,
					}
				}
			}

			return &DataResult{
				Success:    false,
				Timeout:    true,
				Data:       rxTotalBuf,
				DataLength: rxTotalBufCnt,
				Variables:  nil,
			}
		}
	}
}

// checkDeviceAlarm 检查设备报警状态
func (w *worker) checkDeviceAlarm(device *models.Device, deviceCopy *models.Device, variables []models.DeviceProperty) {
	if deviceCopy.CollectTime < 0 {
		return
	}

	alarmStr := make([]string, 0, len(variables))
	copyDeviceProperty := deviceCopy.Type.Properties

	for _, temp := range variables {
		val, ok := temp.Value.(float64)
		if !ok || !temp.IsAlarm {
			continue
		}

		for _, tempDeviceProperty := range copyDeviceProperty {
			if tempDeviceProperty.Name != temp.Name {
				continue
			}

			copyVal, ok := tempDeviceProperty.Value.(float64)
			if !ok {
				continue
			}

			// 检查读数是否异常（比上次小）
			if val < copyVal {
				alarmStr = append(alarmStr, fmt.Sprintf("%s:%.2f 读数异常！比上次读数[%.2f]要小!", temp.Name, val, copyVal))
			}

			// 检查是否超出预警值
			if temp.Threshold > 0 {
				plusVal := copyVal + temp.Threshold
				if val > plusVal {
					alarmStr = append(alarmStr, fmt.Sprintf("%s:%.2f 读数异常！读数超出预警值!", temp.Name, val))
				}
			}
		}
	}

	// 设置报警状态
	if len(alarmStr) > 0 {
		device.AlarmStatus = true
		device.AlarmReason = strings.Join(alarmStr, ",")
		device.AlarmTime = time.Now().Unix()
		device.AlarmTotal += 1
	} else {
		device.AlarmStatus = false
		device.AlarmReason = ""
		device.AlarmTime = 0
	}
}

// calculateAutoValues 计算自动计算属性
func (w *worker) calculateAutoValues(device *models.Device, deviceCopy *models.Device) {
	for k, p := range device.Type.Properties {
		if !p.AutoCalc {
			continue
		}

		for _, cp := range deviceCopy.Type.Properties {
			if cp.Name != p.Name {
				continue
			}

			pv, ok := p.Value.(float64)
			if !ok {
				continue
			}

			cpv, ok := cp.Value.(float64)
			if !ok {
				continue
			}

			if cpv > 0 {
				device.Type.Properties[k].Used = pv - cpv
			} else if device.InitialVal > 0 {
				device.Type.Properties[k].Used = pv - device.InitialVal
			} else {
				device.Type.Properties[k].Used = 0
			}
		}
	}
}

// processSuccessfulCollection 处理成功的数据采集
func (w *worker) processSuccessfulCollection(ctx *DeviceCollectContext, result *DataResult) {
	device := ctx.Device
	device.CollectTime = time.Now().Unix()
	device.Online = true
	device.CollectTotal += 1
	device.CollectSuccess += 1

	// 检查报警
	w.checkDeviceAlarm(device, ctx.DeviceCopy, result.Variables)

	// 计算自动值
	w.calculateAutoValues(device, ctx.DeviceCopy)

	// 添加 original_data 属性
	if len(result.Data) > 0 {
		device.Type.Properties = append(device.Type.Properties, models.DeviceProperty{
			Name:     "original_data",
			Value:    fmt.Sprintf("% 2X", result.Data),
			Reported: true,
		})
	}

	// 保存设备状态
	w.deviceRepository.Save(device)

	if result.Timeout {
		logMsg := fmt.Sprintf("从设备[%s]接收数据后，超时，但正确解析数据", device.Name)
		if ctx.IsCommand {
			zap.S().Infof("RPC%s", logMsg)
		} else {
			zap.S().Info(logMsg)
		}
	} else {
		logMsg := fmt.Sprintf("从设备[%s]接收数据后，正确解析数据", device.Name)
		if ctx.IsCommand {
			zap.S().Infof("RPC%s", logMsg)
		} else {
			zap.S().Info(logMsg)
		}
	}
}

// createDeviceCopy 创建设备深拷贝
func (w *worker) createDeviceCopy(device models.Device) models.Device {
	tempBuff, _ := json.Marshal(device)
	deviceCopy := models.Device{}
	_ = json.Unmarshal(tempBuff, &deviceCopy)
	return deviceCopy
}

// setupDriver 设置设备驱动
func (w *worker) setupDriver(device *models.Device) (string, error) {
	driverName := device.Type.Driver
	if len(driverName) <= 0 {
		find, err := w.deviceTypeRepository.Find(device.Type.Name)
		if err != nil {
			return "", fmt.Errorf("找不到设备类型！%v", err)
		}
		driverName = find.Driver
	}

	w.Runner.Open(driverName)
	return driverName, nil
}

// setupCollector 设置采集器连接
// 
// 连接管理策略：
//   - Serial/TcpClient/MQTT：由 Worker 管理连接的打开和关闭
//   - TcpServer：连接由 zdm.go 的 listener 管理，Worker 只使用已建立的连接
func (w *worker) setupCollector(device *models.Device) func() {
	collectorType := w.Collector.GetName()
	
	// TcpServer 类型：连接由 listener 管理，Worker 不打开/关闭
	if collectorType == "TcpServer" {
		// TcpServer 会在内部查找设备的在线连接并使用
		// 如果设备不在线，采集会失败（这是正常的）
		return func() {} // 不需要关闭连接
	}
	
	// 其他类型：根据配置决定是否每次打开
	if w.config.Serial.OpenEveryTime {
		w.Collector.Open(device)
		return func() { w.Collector.Close() }
	}
	
	// 不是每次都打开，连接在 Worker.Start() 时已打开
	return func() {}
}

func (w *worker) CollectExecutor(task any) {
	device, ok := task.(models.Device)
	if !ok {
		zap.S().Error("数据采集执行器无法转换Device参数")
		return
	}

	jsonData, _ := json.Marshal(device)
	zap.S().Info("开始采集设备: ", string(jsonData))

	// 设置驱动
	driverName, err := w.setupDriver(&device)
	if err != nil {
		zap.S().Error("设置驱动失败: ", err)
		return
	}
	defer w.Runner.Close()

	// 设置采集器连接
	closeCollector := w.setupCollector(&device)
	defer closeCollector()

	// 创建设备副本
	deviceCopy := w.createDeviceCopy(device)

	// 创建采集上下文
	ctx := &DeviceCollectContext{
		Device:     &device,
		DeviceCopy: &deviceCopy,
		DriverName: driverName,
		IsCommand:  false,
	}

	// 数据读取，超过30次，自动退出
	for step := 0; step < 30; step++ {
		ctx.Step = step
		data, result, continued := w.Runner.GenerateGetRealVariables(device.Address, step)

		if result {
			// 数据发送
			zap.S().Infof("向设备[%s]发送数据[%d:%X]", device.Name, len(data), data)
			w.Collector.Write(data)

			// 接收数据
			dataResult := w.receiveData(ctx)

			if dataResult.Success {
				// 处理成功的采集
				w.processSuccessfulCollection(ctx, dataResult)
			} else if dataResult.Timeout {
				// 处理超时情况
				w.handleCollectTimeout(&device, &deviceCopy)
				break
			}
		}

		if !result && !continued {
			w.handleCollectFailure(&device, "未上报读数, 解析失败")
			break
		}

		if !continued {
			break
		}

		time.Sleep(time.Duration(w.Collector.GetInterval()) * time.Millisecond)
	}

	w.eventBus.Publish(event.Event{
		Name: "DeviceCollectFinish",
		Data: device,
	})

	zap.S().Info("结束采集设备:", device.Name)
}

// handleCollectTimeout 处理采集超时
func (w *worker) handleCollectTimeout(device *models.Device, deviceCopy *models.Device) {
	zap.S().Error("超时退出")
	device.AlarmStatus = true
	device.AlarmTime = time.Now().Unix()
	device.AlarmReason = "未上报读数，超时退出"
	device.AlarmTotal += 1
	w.deviceRepository.Save(device)

	// 设置设备离线
	deviceCopy.CollectTime = time.Now().Unix()
	deviceCopy.Online = false
	deviceCopy.CollectTotal += 1
	w.deviceRepository.Save(deviceCopy)
}

// handleCollectFailure 处理采集失败
func (w *worker) handleCollectFailure(device *models.Device, reason string) {
	device.AlarmStatus = true
	device.AlarmTime = time.Now().Unix()
	device.AlarmReason = reason
	device.AlarmTotal += 1
	w.deviceRepository.Save(device)
}

func (w *worker) CommandExecutor(task any) {
	commandRequest, ok := task.(CommandRequest)
	if !ok {
		zap.S().Error("RPC执行器无法转换CommandRequest参数")
		return
	}

	// 串口测试
	if commandRequest.Method == "test" {
		w.CommandTest(commandRequest)
		return
	}

	requestParam := commandRequest.Params[0]
	responseParam := w.createInitialResponseParam(requestParam)

	// 查找设备
	device, err := w.deviceRepository.Find(requestParam.DeviceName)
	if err != nil {
		responseParam.Err = "目标设备不存在！请同步网关设备列表！"
		responseParam.CmdStatus = 1
		commandRequest.ResponseParamChan <- responseParam
		return
	}

	// 创建设备副本
	deviceCopy := w.createDeviceCopy(device)

	// 设置驱动
	driverName, err := w.setupDriver(&device)
	if err != nil {
		zap.S().Error("找不到设备类型！", err, device.Type.Name)
		responseParam.Err = "网关找不到设备类型！"
		responseParam.CmdStatus = 1
		commandRequest.ResponseParamChan <- responseParam
		return
	}
	defer w.Runner.Close()

	// 设置采集器连接
	closeCollector := w.setupCollector(&device)
	defer closeCollector()

	// 处理所有命令参数
	w.processCommandRequests(commandRequest, &device, &deviceCopy, driverName, &responseParam)

	commandRequest.ResponseParamChan <- responseParam
	zap.S().Info("结束执行RPC命令:", commandRequest.Method)
}

// createInitialResponseParam 创建初始响应参数
func (w *worker) createInitialResponseParam(requestParam RequestParam) ResponseParam {
	return ResponseParam{
		ClientID:  requestParam.ClientID,
		CmdName:   requestParam.CmdName,
		CmdStatus: 1,
		CmdResult: nil,
		Err:       "",
	}
}

// processCommandRequests 处理命令请求
func (w *worker) processCommandRequests(commandRequest CommandRequest, device *models.Device, deviceCopy *models.Device, driverName string, responseParam *ResponseParam) {
	// 至少有一条参数（命令）
	for _, requestParam := range commandRequest.Params {
		responseParam.CmdName = requestParam.CmdName

		// 重新查找设备（可能在循环中有变化）
		var err error
		*device, err = w.deviceRepository.Find(requestParam.DeviceName)
		if err != nil {
			break
		}

		// 创建命令上下文
		ctx := &DeviceCollectContext{
			Device:        device,
			DeviceCopy:    deviceCopy,
			DriverName:    driverName,
			IsCommand:     true,
			CommandName:   requestParam.CmdName,
			CommandParams: w.marshalCmdParams(requestParam.CmdParams),
		}

		// 执行命令步骤
		if w.executeCommandSteps(ctx, requestParam, responseParam) {
			break // 如果超时或失败，退出循环
		}
	}
}

// executeCommandSteps 执行命令步骤
func (w *worker) executeCommandSteps(ctx *DeviceCollectContext, requestParam RequestParam, responseParam *ResponseParam) bool {
	// 数据读取，超过30次，自动退出
	for step := 0; step < 30; step++ {
		ctx.Step = step
		cmdParams, _ := json.Marshal(&(requestParam.CmdParams))
		data, result, continued := w.Runner.DeviceCustomCmd(ctx.Device.Address, requestParam.CmdName, string(cmdParams), 0)

		if result {
			// 数据发送
			zap.S().Infof("RPC向设备[%s]发送数据[%d:%X]", ctx.Device.Name, len(data), data)
			w.Collector.Write(data)

			// 接收数据
			dataResult := w.receiveData(ctx)

			if dataResult.Success {
				// 处理成功的命令
				w.processSuccessfulCollection(ctx, dataResult)
				w.fillCommandResponse(responseParam, ctx.Device, dataResult.Variables)
				return false // 成功，不退出
			} else if dataResult.Timeout {
				// 处理超时
				w.handleCommandTimeout(responseParam, dataResult)
				return true // 超时，退出
			}
		}

		if !continued {
			break
		}

		time.Sleep(time.Duration(w.Collector.GetInterval()) * time.Millisecond)
	}

	return false
}

// fillCommandResponse 填充命令响应
func (w *worker) fillCommandResponse(responseParam *ResponseParam, device *models.Device, variables []models.DeviceProperty) {
	responseParam.CmdStatus = 0
	responseParam.DeviceName = device.Name
	tempMap := make(map[string]interface{})
	for _, property := range variables {
		tempMap[property.Name] = property.Value
	}
	responseParam.CmdResult = tempMap
}

// handleCommandTimeout 处理命令超时
func (w *worker) handleCommandTimeout(responseParam *ResponseParam, dataResult *DataResult) {
	if dataResult.DataLength > 0 {
		// 即使超时，如果有数据也尝试解析
		responseParam.CmdStatus = 0
		responseParam.CmdResult = dataResult.Data
	} else {
		responseParam.CmdStatus = 1
		responseParam.CmdResult = nil
	}
}

// marshalCmdParams 序列化命令参数
func (w *worker) marshalCmdParams(cmdParams map[string]interface{}) string {
	data, _ := json.Marshal(&cmdParams)
	return string(data)
}

func (w *worker) CommandTest(commandRequest CommandRequest) {
	// 如果每次都打开，那么在这里打开
	if w.config.Serial.OpenEveryTime {
		zap.S().Error("每次都打开设备，不支持调试!")
		return
	}

	if commandRequest.Method != "test" {
		return
	}

	requestParam := commandRequest.Params[0]

	zap.S().Info("开始执行测试命令:", requestParam.CmdName)

	responseParam := ResponseParam{}

	responseParam.CmdStatus = 0
	responseParam.CmdResult = nil

	// 发送指令
	data, ok := requestParam.CmdParams["param"].([]byte)
	if !ok {
		zap.S().Error("串口测试无法获取参数")
		return
	}
	w.Collector.Write(data)
	zap.S().Infof("发送【测试】数据[%d:%X]", len(data), data)
	rxBuf := make([]byte, 1024)
	rxBufCnt := 0
	rxTotalBuf := make([]byte, 0)
	rxTotalBufCnt := 0

	// 接收指令
	func() {
		for {
			select {
			case rxBuf = <-w.dataChan:
				rxBufCnt = len(rxBuf)
				if rxBufCnt > 0 {
					rxTotalBufCnt += rxBufCnt
					rxTotalBuf = append(rxTotalBuf, rxBuf[:rxBufCnt]...)
					rxBufCnt = 0
					rxBuf = rxBuf[0:0]
				}
			case <-time.After(time.Duration(w.Collector.GetTimeout()) * time.Millisecond):
				responseParam.CmdResult = rxTotalBuf[:rxTotalBufCnt]
				if rxTotalBufCnt == 0 {
					responseParam.CmdStatus = 1
				} else {
					responseParam.CmdStatus = 0
					if rxTotalBufCnt > 0 {
						zap.S().Infof("接收【测试】数据[%d:%X]", rxTotalBufCnt, rxTotalBuf)
					}
				}
				return
			}
		}
	}()

	commandRequest.ResponseParamChan <- responseParam
	zap.S().Infof("结束执行测试命令:", requestParam.CmdName)
}
